package cn.xinfei.xdecision.engine.runner.consumer;

import cn.xinfei.xdecision.common.model.datax.enums.MqDeliveryScene;
import cn.xinfei.xdecision.engine.runner.service.DecisionResultService;
import cn.xinfei.xdecision.engine.runner.service.EngineReentryService;
import cn.xinfei.xdecision.model.VariableMessageVO;
import cn.xinfei.xdecision.pulsar.config.XDecisionMessage;
import cn.xinfei.xdecision.pulsar.consumer.XPulsarConsumerAnnotate;
import com.alibaba.fastjson.JSONObject;
import com.dianping.cat.Cat;
import com.dianping.cat.message.Transaction;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tags;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
@Slf4j
public class XDecisionDataListenerConsumer {

    @Autowired
    private EngineReentryService engineReentryService;

    @Autowired
    private MeterRegistry meterRegistry;

    @Autowired
    private DecisionResultService decisionResultService;


    /**
     * 推荐手动提交ack
     *
     * @param consumer
     * @return
     */
//    @XPulsarConsumerAnnotate(
//            topic = "${xdecision.data.variabletopic}",
//            clazz = String.class,
//            consumerName = "${spring.xpulsar.consumer.consumerName}",
//            subscriptionName = "${spring.xpulsar.consumer.subscriptionName}",
//            batch = false)
    public List<MessageId> consume(Message<String> message, Consumer<String> consumer) {

//        if (null != messages) {
//        Transaction transaction = Cat.newTransaction("xengine", "variabletopic");
//            log.info("Received {} messages in a single batch receive.", messages.size());
//            for (Message<String> message : messages) {
        MessageId messageId = message.getMessageId();
        try {
            String data = message.getValue();
            String key = message.getKey();
            log.info("consume message key={}，收到查询中心征信节点的响应，msg={}", key, data);
            meterRegistry.counter("xengine_consumer_total", Tags.of("topic", message.getTopicName())).increment();
            XDecisionMessage xDecisionMessage = JSONObject.parseObject(data, XDecisionMessage.class);
            String msgId = xDecisionMessage.getMsgId();
            String decisionId = msgId.split("_")[0];
            String variableData = xDecisionMessage.getData();
            VariableMessageVO variableMessage = JSONObject.parseObject(variableData, VariableMessageVO.class);
            String deliveryType = variableMessage.getDeliveryType();
            if (MqDeliveryScene.CREDIT_UPLOAD.getCode().equals(deliveryType)) {
//                        log.info("收到到征信消息，唤醒重入流程.decisionId={}", decisionId);
//                        engineReentryService.reentry(decisionId, "MQ");
            } else if (MqDeliveryScene.CREDIT_QUERY_STOP.getCode().equals(deliveryType)) {
                log.warn("征信查询终止,decisionId={}", decisionId);
//                        decisionResultService.updateDecisionResultCodeById(null, decisionId, EngineResultEnum.CREDIT_PENDING_END.getCode());
            } else if (MqDeliveryScene.CREDIT_VAR_RESULT.getCode().equals(deliveryType)) {
                log.info("收到到征信消息，唤醒重入流程.decisionId={}", decisionId);
                //todo shiyusen
//                        engineReentryService.reentry(decisionId, "MQ");
            } else {
                log.warn("征信查询变量响应报文,decisionId={}", decisionId);
            }
        } catch (Exception e) {
            log.error("消费消息异常,msg={}", e.getMessage(), e);
            Cat.logError(e);
            meterRegistry.counter("xengine_consumer_Exception_total", Tags.of("topic", message.getTopicName())).increment();
        } finally {
            try {
                //此消息永远提交ack，有异步任务会兜底
                consumer.acknowledge(messageId);
            } catch (PulsarClientException pulsarClientException) {
                pulsarClientException.printStackTrace();
            }
            MDC.clear();
        }
//            }
//        transaction.complete();
//        }
        return null;
    }
}
